package com.nulldev.util.net.tcp;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.regex.Pattern;

import com.nulldev.util.encoding.compression.smartCompression;
import com.nulldev.util.io.IOUtils;
import com.nulldev.util.java.JVMUtil;
import com.nulldev.util.java.Variables;
import com.nulldev.util.java.swing.extSwing.eCrashDialog;
import com.nulldev.util.logging.Logger;
import com.nulldev.util.logging.LoggerFactory;
import com.nulldev.util.math.MathUtil;
import com.nulldev.util.net.packets.NetInfo;
import com.nulldev.util.net.packets.NetPacket;
import com.nulldev.util.net.packets.PacketMode;
import com.nulldev.util.other.ArrayUtils;
import com.nulldev.util.other.ByteUtils;
import com.nulldev.util.string.StringsUtil;
import com.nulldev.util.threading.v3.*;

abstract public class TCPCore {
	static private final byte[] empty = new byte[0];
	public final Logger loggerF = LoggerFactory.getLogger(true, false);
	final String category;
	private final String name;
	private final Socket socket;
	//final LargeDataInputStream input;
	final DataInputStream input;
	//final LargeDataOutputStream output; //DataOutputStream output;
	final DataOutputStream output;
	final Object outputLock = new Object();

	final ArrayBlockingQueue<String> sends = new ArrayBlockingQueue(1024, true);
	Thread writeThread;
	public volatile boolean closed;
	byte[] data = empty;

	public TCPCore(String category, String name, final Socket socket) throws IOException {
		this.category = category;
		this.name = name;
		this.socket = socket;
		//this.loggerF.traceEnabled = true;

		try {
			input = new DataInputStream(socket.getInputStream());
			output = new DataOutputStream(socket.getOutputStream());
			//input = new LargeDataInputStream(socket.getInputStream());
			//output = new LargeDataOutputStream(socket.getOutputStream()); //new DataOutputStream(socket.getOutputStream());
		} catch (IOException ex) {
			throw new IOException("Error opening socket streams.", ex);
		}
	}
	
	private boolean wantsToDisconnect;
	public void requestDisconnect() {
		this.wantsToDisconnect = true;
		int i = 3600;
		while (i > 0) {
			try {
				final boolean closed = close();
				if (closed) {
					break;
				}
				Thread.sleep(50L);
				i -= 50;
			}catch (Exception e) {
			}
		}
	}
	
	public Socket getSocket() {
		return this.socket;
	}

	public void start () {
		final Thread readThread = new Thread(name + "Read") {
			public void run () {
				try {
					while (!closed) {
						String message = "";
						try {
							message = input.readUTF();
						}catch (Exception e) {
							try {
								if (!(e instanceof EOFException) && 
										(!(e instanceof SocketException) && !e.getMessage().equals("Socket closed"))) {
									loggerF.error("read() -> Failed to get UTF8 string, falling back...", e);
								}
								message = input.readLine();
							}catch (Exception ea) {
								if (!JVMUtil.isArg("--network.logging") && !(ea instanceof SocketException) && (!(ea instanceof EOFException))) {
									final eCrashDialog crash = eCrashDialog.main(ea);
									crash.setTitle("nullUtil: Networking: Crash Dialog");
									crash.setMessageTitle("Why is networking so hard?");
									crash.waitForOK();
								}
							}
						}
						if (message == null || closed) break;
						
						long readTime = System.currentTimeMillis();
						
						String event, payload;
						int index = message.indexOf(" ");
						if (index != -1) {
							event = message.substring(0, index).trim();
							payload = message.substring(index + 1).trim();
						} else {
							event = message.trim();
							payload = "";
						}

						//int dataLength = TCPUtil.readInt(input);
						final long dataLength = TCPUtil.readLong(input);
						if (dataLength > 0) {
							if (closed) break;
							// TODO: Allow LONG reading.
							if (data.length < dataLength) data = new byte[(int) dataLength];
							int offset = 0, remaining = (int) dataLength;
							while (true) {
								int count = input.read(data, offset, remaining);
								if (count == -1 || closed) break;
								remaining -= count;
								if (remaining == 0) break;
								offset += count;
							}
						}
						
						readTime = System.currentTimeMillis() - readTime;

						if (loggerF.traceEnabled) loggerF.trace(category, "Received: " + event + ", " + payload + (dataLength > 0 ? ", " + dataLength : ""));
						try {
							receive(event.replaceAll(Pattern.quote("%space"), " "), payload, data, dataLength, 
									readTime);
						} catch (Throwable ex) {
							if (loggerF.errorEnabled) loggerF.error(category, "Error processing message: " + (message.length() < 320 ? message :
								"size=" + message.length()), ex);
							break;
						}
					}
				} catch (EOFException ex) {
					if (loggerF.traceEnabled) loggerF.trace(category, "Connection has closed.", ex);
				} catch (IOException ex) {
					if (!closed) {
						if (ex.getMessage() != null && ex.getMessage().contains("Connection reset")) {
							if (loggerF.traceEnabled) loggerF.trace(category, "Client connection reset.", ex);
						} else {
							if (loggerF.errorEnabled) loggerF.error(category, "Error reading from connection.", ex);
						}
					}
				} catch (Exception ex) {
					if (loggerF.errorEnabled) loggerF.error(category, "Error occured.", ex);
				} finally {
					close();
					if (loggerF.traceEnabled) loggerF.trace(category, "Read thread stopped.");
				}
			}
		};
		if (this.PREFER_THREAD_API) {
			ThreadDispatcher.instance().addTask(
					RunnableWorker.fromRunnable(readThread, "$TCP.readThread"));
		}else {
			readThread.start();
		}

		this.writeThread = new Thread(name + "Write") {
			public void run () {
				try {
					while (!closed) {
						try {
							String take = sends.take();
							if (!take.endsWith(JVMUtil.LINE_SEP_BASIC)) {
								take = take + JVMUtil.LINE_SEP_BASIC;
							}
							sendBlocking(take, null, 0, 0);
						} catch (InterruptedException ignored) {
						}
					}
				} finally {
					close();
					if (loggerF.traceEnabled) loggerF.trace(category, "Write thread stopped.");
				}
			}
		};
		if (this.PREFER_THREAD_API) {
			ThreadDispatcher.instance().addTask(
					RunnableWorker.fromRunnable(this.writeThread, "$TCP.writeThread"));
		}else {
			this.writeThread.start();
		}
	}
	
	public boolean PREFER_THREAD_API;

	/** Sends the string without waiting for the send to complete. */
	public void send (String message) {
		if (loggerF.traceEnabled) loggerF.trace(category, "Queued: " + message);
		sends.add(message);
	}
	
	public void send(final NetPacket packet) {
		sends.add(packet.toString());
	}
	
	public boolean sendBlocking(final NetPacket packet) {
		return this.sendBlocking(packet, PacketMode.defaultMode);
	}
	
	public boolean sendBlocking(final NetPacket packet, final PacketMode mode) {
		return this.sendBlocking(packet, mode, null);
	}
	
	public boolean sendBlocking(final NetPacket packet, final NetInfo info) {
		return this.sendBlocking(packet, PacketMode.defaultMode, info);
	}
	
	public boolean sendBlocking(final NetPacket packet, final PacketMode mode, final NetInfo netInfo) {
		final String text = packet.toString(mode, null);
		if (text == null && mode == PacketMode.STANDARD_V2_FULLY_ENCRYPTED) {
			return sendBlocking(packet.getHeader(mode, netInfo), packet.autoReadData(mode, netInfo));
		}else if (text == null && mode == PacketMode.STANDARD_V2_COMPRESSED) {
			final byte[] d = packet.autoReadData(mode, netInfo);
			//JVMUtil.async_error("Compressed using " + smartCompression.compressMethodFromID(smartCompression.getMethod(d)) + ".");
			return sendBlocking(packet.getHeader(mode, netInfo), ByteUtils.toBase64(d).getBytes());
		}
		if (text.getBytes().length >= 62000 || packet.getData().length >= 62000 || packet.preferBytePayload()) {
			return sendBlocking(packet.getHeader(mode, netInfo), packet.autoReadData(mode, netInfo));
		}
		return sendBlocking(text);
	}

	/** Sends the string, blocking until sending is complete.
	 * @return false if the connection is closed or the send failed (which closes the connection). */
	public boolean sendBlocking (String message) {
		return sendBlocking(message, null, 0, 0);
	}

	/** @see #sendBlocking(String, byte[], int, int) */
	public boolean sendBlocking (String message, byte[] bytes) {
		return sendBlocking(message, bytes, 0, bytes.length);
	}

	/** Sends the string, blocking until sending is complete.
	 * @return false if the connection is closed or the send failed (which closes the connection). */
	public boolean sendBlocking(String message, byte[] bytes, int offset, int count) {
		if (message == null) {
			loggerF.error("Failed to send message. [messageNull=" + (message == null) + ",bytesNull=" + (bytes == null) + "]");
			return false;
		}
		message = message.replaceAll(Pattern.quote(" "), "%space");
		if (closed) return false;
		try {
			synchronized (outputLock) {
				if (loggerF.traceEnabled) loggerF.trace(category, "Sent: " + message + (count > 0 ? ", " + count : ""));
//				if (message.getBytes(CharsetUtil.getCharset()).length > 65500) {
//					final byte[][] data = ByteUtils.divideArray(message.getBytes(CharsetUtil.getCharset()), 32000);
//					for (final byte[] dat : data) {
//						output.writeUTF(new String(dat, CharsetUtil.getCharset()));
//					}
//				}else {
//					output.writeBytes(message);
//				}
				if (closed) return false;
				try {
					output.writeUTF(message);
					output.flush();
					//TCPUtil.writeInt(count, output);
					TCPUtil.writeLong(count, output);
					output.flush();
					output.write(bytes, offset, count);
					output.flush();

					
					if (wantsToDisconnect) {
						close();
						wantsToDisconnect = false;
					}
				}catch (Exception e) {
					loggerF.error("Peer disconnected before being able to write.", e);
				}
			}
			return true;
		} catch (Exception ex) {
			if (loggerF.errorEnabled && !closed) loggerF.error(category, "Error writing to connection.", ex);
			close();
			return false;
		}
	}

	abstract public void receive(final String event, final String payload, 
			final byte[] bytes, final long count, final long readTime);

	public boolean close () {
		if (loggerF.traceEnabled && !closed) loggerF.trace(category, "Client disconnected.");
		closed = true;
		if (writeThread != null) writeThread.interrupt();
		IOUtils.closeQuietly(output);
		IOUtils.closeQuietly(input);
		IOUtils.closeQuietly(socket);
		return true;
	}
	
	public Logger getLogger() {
		return this.loggerF;
	}

	public String format() {
		return this.socket.getInetAddress().getHostAddress() + ":" + this.socket.getPort();
	}
	
	//System.out.println(Variables.length(message));
//	if (message != null) {
//		if (!message.startsWith("__p2c")) {
//			final List<String> split = StringsUtil.splitEqually(message, 32000);
//			if ((split.size() == 1 || message.startsWith("__part[")) && message.length() < 64000) {
//				final String message0 = message;
//				//count = message0.getBytes().length;
//				output.writeUTF(message0);
//				TCPUtil.writeInt(count, output);
//				//output.write(bytes, offset, count);
//				output.flush();
//			}else {
//				final String id = StringsUtil.randomString(MathUtil.randInt(28, 32));
//				for (int i = 0; i < split.size(); i++) {
//					if (i + 1 == split.size()) {
//						final String p = "__part[" + id + "]data{" + split.get(i) + "}]}end";
//						sendBlocking(p);
//					}else {
//						final String p = "__part[" + id + "]data{" + split.get(i) + "}]}ep";
//						sendBlocking(p);
//					}
//				}
//			}
//		}else {
//			output.writeUTF(message);
//			TCPUtil.writeInt(count, output);
//			output.write(bytes, offset, count);
//			output.flush();
//		}
//	}else {
//	}
	
//	if (message != null && message.getBytes().length > 64000) {
//		//System.out.println("PROCESSING LARGE MESSAGE: " + message.getBytes().length);
//		bytes = message.getBytes();
//		count = message.getBytes().length;
//		TCPUtil.writeInt(count, output);
//		//output.write(bytes, offset, count);
//		socket.getOutputStream().write(bytes);
//		output.flush();
//	}else {
////		if (message != null) {
////			output.writeUTF(message);
////		}else {
////			output.writeUTF(message);
////			TCPUtil.writeInt(count, output);
////			output.write(bytes, offset, count);
////		}
//		output.writeUTF(message);
//		TCPUtil.writeInt(count, output);
//		output.write(bytes, offset, count);
//		output.flush();
//	}
//	output.writeUTF(message);
//	TCPUtil.writeInt(count, output);
//	output.write(bytes, offset, count);
//	output.flush();
}
